package org.example

import org.apache.spark.{SparkConf, SparkContext}

object PopularTeacher {
  def main(args: Array[String]): Unit ={
    val conf = new SparkConf().setAppName("PopularTeacher").setMaster("local[*]")
    val sc = new SparkContext(conf)

    val inputRDD = sc.textFile("teacher.log")

    val teacherRDD = inputRDD.map(line => line.split("/").last)

    val teacherCountRDD = teacherRDD.map(teacher => (teacher,1)).reduceByKey(_ + _)

    val sortedTeacherCountRDD = teacherCountRDD.sortBy(_._2,ascending = false)

    val mostPopularTeacher = sortedTeacherCountRDD.first()
    println(s"the most popular teacher is :${mostPopularTeacher._1},count is ${mostPopularTeacher._2}")
//    val mostPopularTeacherCount = sortedTeacherCountRDD.collect()
//    mostPopularTeacherCount.foreach{case(teacher, count) => println(s"the most popular teacher is :${teacher},count is ${count}")}
//    println(s"Teacher ${mostPopularTeacher},count:${mostPopularTeacherCount.mkString("Array(", ", ", ")")}")
//    sortedTeacherCountRDD.collect().foreach{case(teacher, count) => println(s"Teacher ${mostPopularTeacher},count:${mostPopularTeacherCount}")}
//
//    println(s"the most popular teacher is :${mostPopularTeacher._1}")
    sc.stop()
  }
}
